package week7.jdbc;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class InsertDemo {

    private static final int SIZE = 10 * 100000;

    public static void main(String[] args) {
        test2();
    }

    public static void test1() {
        System.out.println("单连接批量入库100w");
        doInsertTest("shopping", SIZE, false, DbUtil::executeInsertBatch);
    }

    public static void test2() {
        System.out.println("多连接原生jdbc并发批量入库100w");
        String dataSourceName = "shopping";
        multiThreadInsert(dataSourceName, 500, 4);
    }

    public static void test3() {
        System.out.println("连接池jdbc并发批量入库100w");
        String dataSourceName = "shopping-2";
        multiThreadInsert(dataSourceName, 500, 4);
    }

    private static void multiThreadInsert(String dataSourceName, int batchSize, int threadPoolSize) {
        doInsertTest(dataSourceName, SIZE, false, (sql, list) -> {
            List<List<List<Object>>> partition = partition(list, batchSize);

            ExecutorService service = Executors.newFixedThreadPool(threadPoolSize);

            CountDownLatch countDownLatch = new CountDownLatch(partition.size());

            for (List<List<Object>> l : partition) {
                service.submit(() -> {
                    try {
                        DbUtil.use(dataSourceName, () -> {
                            DbUtil.executeInsertBatch(sql, l);
                        });
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        countDownLatch.countDown();
                    }

                });
            }

            try {
                countDownLatch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            service.shutdown();

        });
    }

    public static void doInsertTest(String dataSourtName, int size, boolean containId, BiConsumer<String, List<List<Object>>> consumer) {

        DbUtil.use(dataSourtName, () -> {

            //用户
            List<Map<String, Object>> userList = DbUtil.executeQuery("select * from user");

            //产品
            List<Map<String, Object>> productList = DbUtil.executeQuery("select * from product");

            //随机构建100w条数据
            int num = size;

            List<List<Object>> lists = generateTestData(num, containId, userList, productList);

            //截断表
            DbUtil.executeUpdate("TRUNCATE TABLE orders");

            System.out.println("insert rows start size====" + num + ",orders count====" + getOrdersCount());

            String sql = getInsertOrderSql(containId);

            long start = System.currentTimeMillis();

            consumer.accept(sql, lists);

            long end = System.currentTimeMillis();

            System.out.println("insert rows finish,orders count==== " + getOrdersCount() + ",cost====" + (end - start));

        });

    }

    public static int getOrdersCount() {
        List<Map<String, Object>> res = DbUtil.executeQuery("select count(1) count from orders");

        Object count = res.get(0).get("count");

        return count == null ? 0 : Integer.parseInt(String.valueOf(count));
    }


    public static String getInsertOrderSql(boolean containId) {
        if (containId) {
            return "insert into orders(order_id,user_id,product_id,payment,payment_type,postage,status,payment_time,create_time,update_time) values(?,?,?,?,?,?,?,?,?,?)";
        } else {
            return "insert into orders(user_id,product_id,payment,payment_type,postage,status,payment_time,create_time,update_time) values(?,?,?,?,?,?,?,?,?)";
        }
    }

    public static List<List<Object>> generateTestData(int size, boolean containId, List<Map<String, Object>> userList, List<Map<String, Object>> productList) {

        List<List<Object>> result = new ArrayList<>();

        for (int i = 0; i < size; i++) {
            int u = new Random().nextInt(userList.size());
            int p = new Random().nextInt(productList.size());

            Map<String, Object> uo = userList.get(u);
            Map<String, Object> po = productList.get(u);

            List<Object> row = new ArrayList<>();

            //user_id
            //product_id
            //payment
            //payment_type
            //postage
            //status
            //payment_time
            //send_time
            //end_time
            //close_time
            //create_time
            //update_time

            if (containId) {
                row.add(i + 1);
            }
            //user_id
            row.add(uo.get("user_id"));
            //product_id
            row.add(po.get("product_id"));
            //payment 取价格
            row.add(po.get("price"));
            //payment_type
            row.add(1);
            //postage
            row.add(8);
            //status
            row.add(20);
            //payment_time
            row.add(new Date());
            //create_time
            row.add(new Date());
            //update_time
            row.add(new Date());

            result.add(row);

        }

        return result;

    }

    private static <T> List<List<T>> partition(final List<T> list, final int size) {
        Integer limit = (list.size() + size - 1) / size;
        List<List<T>> mglist = new ArrayList<List<T>>();
        Stream.iterate(0, n -> n + 1).limit(limit).forEach(i -> {
            mglist.add(list.stream().skip(i * size).limit(size).collect(Collectors.toList()));
        });
        return mglist;
    }

}